package beautiful.butterfly.schedule.task_manager;

import beautiful.butterfly.schedule.MultipleTaskDeal;
import beautiful.butterfly.schedule.SingleTaskDeal;
import beautiful.butterfly.schedule.TaskDeal;
import beautiful.butterfly.schedule.TaskItemAndParameter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Array;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


public class NotSleepProcessor<Task> implements IProcessor, Runnable {

    private static transient Logger logger = LoggerFactory.getLogger(NotSleepProcessor.class);
    protected AbstractScheduleManager abstractScheduleManager;
    protected TaskDeal<Task> taskDeal;
    protected List<Task> taskList = new CopyOnWriteArrayList<Task>();
    protected List<Object> runningTaskList = new CopyOnWriteArrayList<Object>();
    protected List<Task> maybeRepeatTaskList = new CopyOnWriteArrayList<Task>();
    List<Thread> threadList = new CopyOnWriteArrayList<Thread>();
    TaskType taskType;
    Comparator<Task> taskComparator;
    Statistics statistics;
    Lock fetchSingleTaskLock = new ReentrantLock();
    Lock fetchMultipleTaskLock = new ReentrantLock();
    Lock lockLoadData = new ReentrantLock();

    volatile boolean isMultipleTask = false;


    volatile boolean isStopSchedule = false;
    volatile boolean isSleeping = false;


    public NotSleepProcessor(AbstractScheduleManager abstractScheduleManager,
                             TaskDeal<Task> taskDeal, Statistics statistics) throws Exception {
        this.abstractScheduleManager = abstractScheduleManager;
        this.statistics = statistics;
        this.taskType = this.abstractScheduleManager.getTaskType();
        this.taskDeal = taskDeal;
        this.taskComparator = new TaskComparator(this.taskDeal.getComparator());
        if (this.taskDeal instanceof SingleTaskDeal<?>) {
            if (taskType.getMaxExecuteNumber() > 1) {
                taskType.setMaxExecuteNumber(1);
            }
            isMultipleTask = false;
        } else {
            isMultipleTask = true;
        }
        if (taskType.getFetchDataNumber() > taskType.getThreadNumber() * 10) {
            logger.warn("fetchnum >= threadnum *10 ");
        }
        for (int i = 0; i < taskType.getThreadNumber(); i++) {
            this.startThread(i);//N个线程
        }
    }


    public void stopSchedule() throws Exception {
        this.isStopSchedule = true;
        this.taskList.clear();
    }

    private void startThread(int index) {
        Thread thread = new Thread(this);//为什么传入的是this
        threadList.add(thread);
        String threadName = this.abstractScheduleManager.getTaskHandler().getTaskType() + "-" + this.abstractScheduleManager.getCurrentSerialNumber() + "-exe" + index;
        thread.setName(threadName);
        thread.start();
    }

    @SuppressWarnings("unchecked")
    protected boolean isDealing(Task task) {
        if (this.maybeRepeatTaskList.size() == 0) {
            return false;
        }
        Task[] tasks = (Task[]) this.maybeRepeatTaskList.toArray();
        for (int i = 0; i < tasks.length; i++) {
            if (this.taskComparator.compare(task, tasks[i]) == 0) {
                this.maybeRepeatTaskList.remove(tasks[i]);
                return true;
            }
        }
        return false;
    }


    public Task getSingleDealingTask() {
        fetchSingleTaskLock.lock();
        try {
            Task task = null;
            while (true) {
                if (this.taskList.size() > 0) {
                    task = this.taskList.remove(0);
                } else {
                    return null;
                }
                if (this.isDealing(task) == false) {
                    return task;
                }
            }
        } finally {
            fetchSingleTaskLock.unlock();
        }
    }


    @SuppressWarnings("unchecked")
    public Task[] getMultipleTask() {
        fetchMultipleTaskLock.lock();
        try {
            if (this.taskList.size() == 0) {
                return null;
            }
            int size = taskList.size() > taskType.getMaxExecuteNumber() ? taskType.getMaxExecuteNumber() : taskList.size();//min

            List<Task> taskList = new ArrayList<Task>();
            int count = 0;
            Task task = null;
            while (count < size && ((task = this.getSingleDealingTask()) != null)) {
                taskList.add(task);
                count = count + 1;
            }
            if (taskList.size() == 0) {
                return null;
            } else {
                return taskList.toArray((Task[]) Array.newInstance(taskList.get(0).getClass(), 0));
            }
        } finally {
            fetchMultipleTaskLock.unlock();
        }
    }

    public void clearTaskList() {
        this.taskList.clear();
    }

    public boolean taskListIsAllFinshed() {
        return this.taskList.size() == 0 && this.runningTaskList.size() == 0;
    }

    public boolean isSleeping() {
        return this.isSleeping;
    }


    protected int loadTask() {
        lockLoadData.lock();
        try {
            if (this.taskList.size() > 0 || this.isStopSchedule == true) {
                return this.taskList.size();
            }

            try {
                if (this.taskType.getSleepTimeInterval() > 0) {
                    if (logger.isTraceEnabled()) {
                        logger.trace(""
                                + this.taskType.getSleepTimeInterval());
                    }
                    this.isSleeping = true;
                    Thread.sleep(taskType.getSleepTimeInterval());
                    this.isSleeping = false;


                }
            } catch (Throwable ex) {
                logger.error("", ex);
            }

            putRunningTaskListToMaybeRepeatTaskList();

            try {
                List<TaskItemAndParameter> taskItemAndParameterList = this.abstractScheduleManager.getCurrentTaskItemAndParameterList();

                if (taskItemAndParameterList.size() > 0) {
                    List<TaskItemAndParameter> newTaskItemAndParameterList = new ArrayList<TaskItemAndParameter>();
                    synchronized (taskItemAndParameterList) {
                        for (TaskItemAndParameter taskItemAndParameter : taskItemAndParameterList) {
                            newTaskItemAndParameterList.add(taskItemAndParameter);
                        }
                    }
                    List<Task> taskList = this.taskDeal.getTaskList(
                            taskType.getTaskParameter(),
                            abstractScheduleManager.getTaskHandler()
                                    .getOwnSign(), this.abstractScheduleManager.getTaskItemCount(), taskType.getFetchDataNumber(), newTaskItemAndParameterList
                    );
                    abstractScheduleManager.getTaskHandler().setLastFetchDataTime(new Timestamp(abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()));
                    if (taskList != null) {
                        this.taskList.addAll(taskList);
                    }
                } else {

                }
                addFetchNum(taskList.size(),
                        "TBScheduleProcessor.getTaskList");
                if (taskList.size() <= 0) {

                    if (this.abstractScheduleManager.isContinueWhenData() == true) {
                        if (taskType.getSleepTimeNoData() > 0) {
                            if (logger.isDebugEnabled()) {
                                logger.debug("sleep "
                                        + taskType.getSleepTimeNoData());
                            }
                            this.isSleeping = true;
                            Thread.sleep(taskType.getSleepTimeNoData());
                            this.isSleeping = false;
                        }
                    }
                }
                return this.taskList.size();
            } catch (Throwable ex) {
                logger.error("", ex);
            }
            return 0;
        } finally {
            lockLoadData.unlock();
        }
    }


    @SuppressWarnings("unchecked")
    public void putRunningTaskListToMaybeRepeatTaskList() {
        fetchSingleTaskLock.lock();
        try {
            this.maybeRepeatTaskList.clear();
            if (this.runningTaskList.size() == 0) {
                return;
            }
            Object[] objects = this.runningTaskList.toArray();
            for (int i = 0; i < objects.length; i++) {
                if (this.isMultipleTask == false) {
                    this.maybeRepeatTaskList.add((Task) objects[i]);
                } else {
                    Task[] tasks = (Task[]) objects[i];
                    for (int j = 0; j < tasks.length; j++) {
                        this.maybeRepeatTaskList.add(tasks[j]);
                    }
                }
            }
        } finally {
            fetchSingleTaskLock.unlock();
        }
    }

    /**
     * 具体执行的方法
     */
    @SuppressWarnings("unchecked")
    public void run() {
        long startTime = 0;
        long sequence = 0;
        Object task = null;//可能多个也可能一个task
        while (true) {
            try {
                if (this.isStopSchedule == true) {
                    synchronized (this.threadList) {
                        this.threadList.remove(Thread.currentThread());
                        if (this.threadList.size() == 0) {
                            this.abstractScheduleManager.unRegisterTaskHandler();
                        }
                    }
                    return;
                }

                if (this.isMultipleTask == false) {
                    task = this.getSingleDealingTask();
                } else {
                    task = this.getMultipleTask();
                }
                if (task == null) {
                    this.getTaskSize();
                    continue;
                }

                try {
                    this.runningTaskList.add(task);
                    startTime = abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis();
                    sequence = sequence + 1;
                    if (this.isMultipleTask == false) {
                        if (((SingleTaskDeal<Object>) this.taskDeal).execute(task, abstractScheduleManager.getTaskHandler().getOwnSign()) == true) {
                            addSuccessNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                            - startTime,
                                    "com.taobao.pamirs.schedule.NotSleepProcessor.run");
                        } else {
                            addFailNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                            - startTime,
                                    "com.taobao.pamirs.schedule.NotSleepProcessor.run");
                        }
                    } else {
                        if (((MultipleTaskDeal<Object>) this.taskDeal)
                                .execute((Object[]) task, abstractScheduleManager.getTaskHandler().getOwnSign()) == true) {
                            addSuccessNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                            - startTime,
                                    "com.taobao.pamirs.schedule.NotSleepProcessor.run");
                        } else {
                            addFailNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                            - startTime,
                                    "com.taobao.pamirs.schedule.NotSleepProcessor.run");
                        }
                    }
                } catch (Throwable ex) {
                    if (this.isMultipleTask == false) {
                        addFailNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis() - startTime,
                                "TBScheduleProcessor.run");
                    } else {
                        addFailNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime,
                                "TBScheduleProcessor.run");
                    }
                    logger.error("Task :" + task + " ", ex);
                } finally {
                    this.runningTaskList.remove(task);
                }
            } catch (Throwable e) {
                throw new RuntimeException(e);
                //log.error(e.getMessage(), e);
            }
        }
    }

    public void addFetchNum(long num, String addr) {
        this.statistics.addFetchDataCount(1);
        this.statistics.addFetchDataNum(num);
    }

    public void addSuccessNum(long num, long spendTime, String addr) {
        this.statistics.addDealDataSucess(num);
        this.statistics.addDealSpendTime(spendTime);
    }

    public void addFailNum(long num, long spendTime, String addr) {
        this.statistics.addDealDataFail(num);
        this.statistics.addDealSpendTime(spendTime);
    }

    class TaskComparator implements Comparator<Task> {
        Comparator<Task> comparator;

        public TaskComparator(Comparator<Task> comparator) {
            this.comparator = comparator;
        }

        public int compare(Task leftTask, Task rightTask) {
            statistics.addOtherCompareCount(1);
            return this.comparator.compare(leftTask, rightTask);
        }

        public boolean equals(Object obj) {
            return this.comparator.equals(obj);
        }
    }

}
